/*
 * Copyright 2020-2025 the original author or authors.
 * You cannot use this file unless authorized by the author.
 */
package org.ipig.messaging.kafka.consumer;

import org.ipig.commons.conf.kafka.consumer.KafkaConsumerConf;
import org.ipig.commons.service.ReporterService;
import org.ipig.commons.service.retry.RetryAbleService;
import org.ipig.constants.ResultStatus;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.kafka.clients.consumer.CommitFailedException;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.text.MessageFormat;
import java.time.Duration;
import java.util.Arrays;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

/**
 * AbstractKafkaConsumer
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @version $$Id: AbstractKafkaConsumer.java 3 2020-01-21 09:36:33Z ts <comchnts@163.com> $$
 * @since 1.0
 */
@Slf4j
public abstract class AbstractKafkaConsumer implements KafkaConsumerService {
    private LongAdder allCount = new LongAdder();
    private LongAdder successCount = new LongAdder();
    private LongAdder repeatCount = new LongAdder();
    private final int retryMaxQty = 3;
    private final long waitMillis = 500;
    private volatile boolean flag = true;

    /**
     * 汇报器服务
     */
    @Setter
    @Getter
    private ReporterService<String, Void> reporterService;
    /**
     * Consumer
     */
    private KafkaConsumer kafkaConsumer;
    /**
     * kafka消息配置
     */
    @Setter
    @Getter
    private KafkaConsumerConf conf;

    AbstractKafkaConsumer() {
    }

    /**
     * process
     *
     * @param records
     */
    public abstract ResultStatus process(ConsumerRecords<String, String> records);

    private ResultStatus doProcess(ConsumerRecords<String, String> records) {
        return this.process(records);
    }

    @Override
    public Void execute() {
        if (flag) {
            throw new RuntimeException("请先进行初始化，调用init()方法");
        }
        for (; ; ) {
            //poll(final long timeoutMs)
            // 如果缓冲区中没有数据，则在轮询中等待的时间（以毫秒为单位
            //如果为0，则立即返回缓冲区中当前可用的任何记录，否则返回空。
            //自2.0以来已弃用。使用@link poll（duration），它不会阻止超过等待分区的超时。
            try {
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(300));
                if (records != null && !records.isEmpty()) {
                    int recordCount = records.count();
                    allCount.add(recordCount);
                    new RetryAbleService() {
                        public boolean process() {
                            boolean retryAble = false;
                            ResultStatus resultStatus = doProcess(records);
                            if (resultStatus == ResultStatus.SUCCESS) {
                                retryAble = true;
                                successCount.add(recordCount);
                                try {
                                    kafkaConsumer.commitSync();//手动提交
                                    log.info("Submit successfully : message qty={}", recordCount);
                                } catch (CommitFailedException cfe) {
                                    System.out.println("本次提交失败，offset接下来将再次提交，error message: " + cfe.getMessage());
                                    log.warn("本次提交失败，offset接下来将再次提交，，error message: {}", cfe.getMessage());
                                } catch (Exception ex) {
                                    ex.printStackTrace();
                                }
                            } else {
                                repeatCount.add(recordCount);
                                System.out.println(recordCount + "条消息处理失败，将进行重试处理,请确保业务处理服务的幂等性");
                                log.warn("{}条消息处理失败，将进行重试处理,请确保业务处理服务的幂等性", recordCount);
                            }
                            return retryAble;
                        }

                        @Override
                        public void retryFailed() {
                            System.out.println(retryMaxQty + "次重试之后,消息最终处理失败，为了防止消息丢失，程序将终止");
                            log.warn("{}次重试之后,消息最终处理失败，为了防止消息丢失，程序将终止", retryMaxQty);
                            System.exit(1);
                        }
                    }.retry(retryMaxQty, waitMillis);
                }
            } catch (Throwable ex) {
                ex.printStackTrace();
                log.error(ExceptionUtils.getStackTrace(ex));
            }
        }
    }

    /**
     * init
     */
    public void init() {
        if (flag) {
            synchronized (this) {
                if (conf == null) {
                    throw new RuntimeException("configure the KafkaConsumerConf instance");
                }
                Properties props = new Properties();
                props.put("bootstrap.servers", conf.getBootstrapServers());
                props.put("group.id", conf.getGroupId());
                if (StringUtils.equalsIgnoreCase("true", conf.getAutoCommitEnable())) {
                    props.put("enable.auto.commit", conf.getAutoCommitEnable());
                    props.put("auto.commit.interval.ms", conf.getAutoCommitMs());
                } else {
                    props.put("enable.auto.commit", "false");
                }
                props.put("auto.offset.reset", conf.getAutoOffsetReset());
                props.put("max.poll.records", conf.getMaxPollRecords());
                props.put("key.deserializer", conf.getKeyDeserializer());
                props.put("value.deserializer", conf.getValueDeserializer());
                props.put("session.timeout.ms", conf.getSessionTimeoutMs());
                props.put("max.poll.interval.ms", conf.getMaxPollIntervalMs());
                props.put("request.timeout.ms", conf.getRequestTimeoutMs());
                props.put("fetch.max.wait.ms", conf.getFetchMaxWaitMs());
                props.put("heartbeat.interval.ms", conf.getHeartbeatIntervalMs());
                props.put("connections.max.idle.ms", conf.getConnectionsMaxIdleMs());
                props.put("default.api.timeout.ms", conf.getDefaultApiTimeoutMs());
                props.put("metadata.max.age.ms", conf.getMetadataMaxAgeMs());
                props.put("metrics.sample.window.ms", conf.getMetricsSampleWindowMs());

                Map<String, String> kv = conf.getMap();
                if (kv != null && !kv.isEmpty()) {
                    props.putAll(kv);
                }
                kafkaConsumer = new KafkaConsumer<>(props);
                kafkaConsumer.subscribe(Arrays.asList(conf.getTopicName()));
                log.debug("KafkaConsumer init completed");
                flag = false;
                ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
                scheduledExecutor.scheduleAtFixedRate(() -> {
                    String msg = MessageFormat.format("当前Consumer处理过的总数据量{}【包括重复处理记录】,其中提交offsets的量:{}", allCount.longValue(), successCount.longValue());
                    if (reporterService != null) {
                        reporterService.report(msg);
                    } else {
                        log.info(msg);
                    }
                }, 10, 30, TimeUnit.SECONDS);
            }
        }
    }
}